package org.apache.flink.formats.jsontm;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.DeserializationFormatFactory;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.factories.FactoryUtil;

import java.util.Collections;
import java.util.HashSet;
import java.util.Set;

import static org.apache.flink.formats.json.JsonFormatOptions.*;

/**
 * <ol>
 * date:2022/1/4 editor:yanghongjian
 * <li>创建文档</li>
 * <li></li>
 * </ol>
 * <ol>
 *
 * @author <a href="mailto:12719889@qq.com">YangHongJian</a>
 * @version 3.0
 * @since 1.8
 */
public class JsonTmFormatFactory implements DeserializationFormatFactory {

    public static final ConfigOption<String> ROOT_PATH = ConfigOptions.key("root").stringType().defaultValue("").withDescription("Optional of json root path, null by default.");
    /**
     * 工厂标识符，用来匹配 'format'='json-tm'
     *
     * @return 返回工厂标识符
     */
    @Override
    public String factoryIdentifier() {
        return "json-tm";
    }

    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        return Collections.emptySet();
    }

    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        final Set<ConfigOption<?>> options = new HashSet<>();
        options.add(ROOT_PATH);
        options.add(FAIL_ON_MISSING_FIELD);
        options.add(IGNORE_PARSE_ERRORS);
        options.add(MAP_NULL_KEY_MODE);
        options.add(MAP_NULL_KEY_LITERAL);
        options.add(TIMESTAMP_FORMAT);
        options.add(ENCODE_DECIMAL_AS_PLAIN_NUMBER);
        return options;
    }

    @Override
    public DecodingFormat<DeserializationSchema<RowData>> createDecodingFormat(DynamicTableFactory.Context context, ReadableConfig formatOptions) {
        // 这里实现自定义验证逻辑 或 使用提供的帮助工具类 FactoryUtil
        FactoryUtil.validateFactoryOptions(this, formatOptions);

        boolean failOnMissingField = formatOptions.get(FAIL_ON_MISSING_FIELD);
        boolean ignoreParseErrors = formatOptions.get(IGNORE_PARSE_ERRORS);
        if (ignoreParseErrors && failOnMissingField) {
            throw new ValidationException("fail-on-missing-field 和 ignore-parse-errors 不能同时配置为 true");
        }
        return new JsonTmDecodingFormat(formatOptions);
    }
}
